package com.sjc.zeye.product;

import com.sjc.zeye.product.domain.UserBehavior;
import com.sjc.zeye.product.function.ProductCount;
import com.sjc.zeye.product.function.ProductEventTimeExtractor;
import com.sjc.zeye.product.function.TopHotProduct;
import com.sjc.zeye.product.function.WindowResult;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

public class ZzHotProduct {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 使用事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        env.readTextFile("C:\\resources\\data1.csv")
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String line) throws Exception {
                        String[] fields = line.split(",");
                        return new UserBehavior(Long.parseLong(fields[0].trim()),
                                Long.parseLong(fields[1].trim()), Integer.parseInt(fields[0].trim()),
                                fields[3].trim(),Long.parseLong(fields[0].trim()));
                    }
                }).assignTimestampsAndWatermarks(new ProductEventTimeExtractor())
                .filter(new FilterFunction<UserBehavior>() {
                    @Override
                    public boolean filter(UserBehavior userBehavior) throws Exception {
                        return userBehavior.getBehavior().equals("pv");
                    }
                })
                .keyBy("productId")
                .timeWindow(Time.hours(1),Time.minutes(5))
                .aggregate(new ProductCount(), new WindowResult())
                .keyBy("windowEnd")
                .process(new TopHotProduct(5))
        .print();

        env.execute("ZzHotProduct");
    }
}
